import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

public class spout1 extends BaseRichSpout {

    private static final long serialVersionUID = -9174998944310422274L;
    private SpoutOutputCollector collector;
    private ConsumerConnector consumer;
    private String topic;

    public spout1() {}
    public spout1(String topic) {
        this.topic = topic;
    }
    public void nextTuple() {    }

    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    public void ack(Object msgId) {    }

    public void activate() {
        consumer =kafka.consumer.Consumer.createJavaConsumerConnector(createConsumerConfig());
        Map<String,Integer> topickMap = new HashMap<String, Integer>();
        topickMap.put(topic, 1);
        System.out.println("*********Results********topic:"+topic);

        try{
            Map<String, List<KafkaStream<byte[],byte[]>>>  streamMap=consumer.createMessageStreams(topickMap);
            KafkaStream<byte[],byte[]>stream = streamMap.get(topic).get(0);
            ConsumerIterator<byte[],byte[]> it =stream.iterator();
            while(it.hasNext()){
                String value =new String(it.next().message());
                //System.out.println("storm接收到来自kafka的消息------->" + value);
                String[] words = value.split(",");
                //接收数据
                String book_name=words[1];
                String book_num=words[2].replace("\"", "");
                String book_type=words[3];
                String book_press=words[4];
                //将数据提交到下一个bolt中
                collector.emit(new Values(book_name,book_num,book_type,book_press), value);
            }}catch (Exception e){}
    }

    private static ConsumerConfig createConsumerConfig() {
        Properties props = new Properties();
        // 设置zookeeper的链接地址
        props.put("zookeeper.connect","192.168.128.140:2181,192.168.128.141:2181,192.168.128.142:2181,192.168.128.143:2181");
        // 设置group id
        props.put("group.id", "1");
        // kafka的group 消费记录是保存在zookeeper上的, 但这个信息在zookeeper上不是实时更新的, 需要有个间隔时间更新
        props.put("auto.commit.interval.ms", "1000");
        props.put("zookeeper.session.timeout.ms","10000");
        return new ConsumerConfig(props);
    }

    public void close() {    }

    public void deactivate() {    }

    public void fail(Object msgId) {    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        //声名emit变量
        declarer.declare(new Fields("book_name","book_num","book_type","book_press"));
    }

    public Map<String, Object> getComponentConfiguration() {
        System.out.println("getComponentConfiguration被调用");
        //设置接入的kafka topic
        topic="order_end";
        return null;
    }
}